package com.bianpai.execution;

import com.bianpai.core.Flow;
import com.bianpai.core.FlowContext;
import com.bianpai.core.Node;
import com.bianpai.core.NodeResult;
import com.bianpai.log.FlowLogger;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * 并行执行策略，根据节点依赖关系并行执行
 */
public class ParallelExecutionStrategy implements ExecutionStrategy {
    private int maxThreads;
    private ExecutorService executorService;
    
    public ParallelExecutionStrategy() {
        this(Runtime.getRuntime().availableProcessors());
    }
    
    public ParallelExecutionStrategy(int maxThreads) {
        this.maxThreads = maxThreads;
        this.executorService = Executors.newFixedThreadPool(maxThreads);
    }
    
    @Override
    public boolean execute(Flow flow, FlowContext context, FlowLogger logger) {
        try {
            // 构建节点依赖图
            Map<String, Set<String>> dependencyGraph = buildDependencyGraph(flow);
            
            // 找出没有依赖的节点（入度为0的节点）
            Set<String> nodesWithNoDependencies = findNodesWithNoDependencies(flow, dependencyGraph);
            
            // 已完成的节点
            Set<String> completedNodes = new ConcurrentSkipListSet<>();
            
            // 执行结果
            Map<String, NodeResult> results = new ConcurrentHashMap<>();
            
            // 是否有节点执行失败
            AtomicBoolean hasFailure = new AtomicBoolean(false);
            
            // 当还有节点可以执行且没有失败时，继续执行
            while (!nodesWithNoDependencies.isEmpty() && !hasFailure.get()) {
                // 并行执行当前可执行的节点
                List<CompletableFuture<Void>> futures = new ArrayList<>();
                
                for (String nodeId : nodesWithNoDependencies) {
                    Node node = flow.getNodeById(nodeId);
                    
                    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                        logger.logNodeStart(node, context);
                        
                        try {
                            NodeResult result = node.execute(context);
                            results.put(nodeId, result);
                            context.setNodeResult(nodeId, result);
                            
                            logger.logNodeComplete(node, context, result);
                            
                            if (!result.isSuccess()) {
                                hasFailure.set(true);
                            }
                        } catch (Exception e) {
                            NodeResult result = NodeResult.failure("Node execution error: " + e.getMessage(), e);
                            results.put(nodeId, result);
                            context.setNodeResult(nodeId, result);
                            logger.logNodeError(node, context, e);
                            hasFailure.set(true);
                        }
                        
                        // 标记节点为已完成
                        completedNodes.add(nodeId);
                    }, executorService);
                    
                    futures.add(future);
                }
                
                // 等待当前批次的节点执行完成
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
                
                // 如果有节点执行失败，终止流程
                if (hasFailure.get()) {
                    return false;
                }
                
                // 找出下一批可执行的节点
                nodesWithNoDependencies = findNextExecutableNodes(flow, dependencyGraph, completedNodes);
            }
            
            // 检查是否所有节点都已执行
            return completedNodes.size() == flow.getNodes().size();
        } finally {
            // 不关闭线程池，因为可能会被重复使用
        }
    }
    
    /**
     * 构建节点依赖图
     * @param flow 流程定义
     * @return 依赖图，key为节点ID，value为该节点依赖的节点ID集合
     */
    private Map<String, Set<String>> buildDependencyGraph(Flow flow) {
        Map<String, Set<String>> graph = new HashMap<>();
        
        // 初始化图
        for (Node node : flow.getNodes()) {
            graph.put(node.getId(), new HashSet<>());
        }
        
        // 添加依赖关系
        for (Map.Entry<String, List<String>> entry : flow.getDependencies().entrySet()) {
            String nodeId = entry.getKey();
            List<String> dependencies = entry.getValue();
            
            graph.get(nodeId).addAll(dependencies);
        }
        
        return graph;
    }
    
    /**
     * 找出没有依赖的节点
     * @param flow 流程定义
     * @param dependencyGraph 依赖图
     * @return 没有依赖的节点ID集合
     */
    private Set<String> findNodesWithNoDependencies(Flow flow, Map<String, Set<String>> dependencyGraph) {
        return flow.getNodes().stream()
                .map(Node::getId)
                .filter(nodeId -> dependencyGraph.get(nodeId).isEmpty())
                .collect(Collectors.toSet());
    }
    
    /**
     * 找出下一批可执行的节点
     * @param flow 流程定义
     * @param dependencyGraph 依赖图
     * @param completedNodes 已完成的节点
     * @return 可执行的节点ID集合
     */
    private Set<String> findNextExecutableNodes(Flow flow, Map<String, Set<String>> dependencyGraph, Set<String> completedNodes) {
        return flow.getNodes().stream()
                .map(Node::getId)
                .filter(nodeId -> !completedNodes.contains(nodeId)) // 未完成的节点
                .filter(nodeId -> completedNodes.containsAll(dependencyGraph.get(nodeId))) // 所有依赖都已完成
                .collect(Collectors.toSet());
    }
    
    /**
     * 关闭线程池
     */
    public void shutdown() {
        executorService.shutdown();
    }
}